package com.anli02.kafkaStream;


import com.anli02.entity.dto.UpdateArticleMess;
import com.anli02.entity.enums.HotArticleConstants;
import com.anli02.entity.pojo.ArticleVisitStreamMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.alibaba.fastjson.JSON;

import java.security.Key;
import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {
    ///从某个 Kafka 主题中读取数据并进行处理
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder) {
        //读取数据聚合队列消息-数据消息miss类型
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_AGGREGATION_QUEUE);
        //聚合流式处理
        stream.map((key,value)->
                {
                    UpdateArticleMess mess= JSON.parseObject(value,UpdateArticleMess.class);
                    //重置消息的key:1234343434(musicid)   和  value: likes:1(typeAndAdd)
                    String musicid = mess.getMusicid();
                    String typeAndAdd =mess.getType().name()+mess.getAdd();
                    return new KeyValue<>(musicid, typeAndAdd);
                }///第一层数据转换stream
                ).groupBy((key,value)->key)//根据id分组
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//时间窗口堆积处理
                        .aggregate(new Initializer<String>()
                                   {///    // 执行聚合操作，初始化为 "0"
                                       @Override
                                       public String apply() {
                                           return "view:0,comment:0,like:0";
                                       }
                                   },new Aggregator<String,String,String>()
                                   {
                                       /**
                                        * 真正的聚合操作，返回值是消息的value
                                        */
                                       @Override
                                       public String apply(String key, String value, String aggValue) {
                                           //无值直接返回
                                           if(StringUtils.isBlank(value))
                                           {
                                               return aggValue;
                                           }
                                           ///"view:0,comment:0,like:0"
                                           /// likes:1(typeAndAdd)
                                           String[] aggAry = aggValue.split(",");
                                           int com=0,lik=0,vie=0;//评论 点赞 浏览
                                           for (String item : aggAry) {
                                               //拿出对应的点赞 浏览 评论次数
                                               if(item.startsWith("comment"))
                                               {
                                                   com=Integer.parseInt(item.split(":")[1]);
                                               }else if(item.startsWith("like"))
                                               {
                                                   lik=Integer.parseInt(item.split(":")[1]);
                                               }else if(item.startsWith("view"))
                                               {
                                                   vie=Integer.parseInt(item.split(":")[1]);
                                               }

                                           }
                                           ///累加操作
                                           String[] valAry = value.split(":");
                                           if(valAry[0].equals("view"))
                                           {
                                               vie+=Integer.parseInt(valAry[1]);
                                           }else if(valAry[0].equals("comment"))
                                           {
                                               com+=Integer.parseInt(valAry[1]);
                                           }else if(valAry[0].equals("like"))
                                           {
                                               lik+=Integer.parseInt(valAry[1]);
                                           }
                                           // 进行统计
                                           String formatStr = String.format("COMMENT:%d,LIKES:%d,VIEWS:%d",  com, lik, vie);
                                           System.out.println("文章的id:"+key);
                                           System.out.println("当前时间窗口内的消息处理结果："+formatStr);

                                           return formatStr;
                                       }
                                   }, Materialized.as("hot-article-stream-aggregation-store"))// 指定状态存储
                ///Kafka Streams 会使用这个名称为你的聚合操作创建一个状态存储，并将结果存储在本地（或远程）供后续查询和使用。
                //使用 Materialized.as() 可以创建命名的存储。如果你想在流处理应用程序中查询聚合结果，可以通过状态存储的名称找到对应的结果。
                                .toStream().map((key,value)->{

                            return new KeyValue<>(key.key().toString(),formatObj(key.key().toString(),value));
                        }).to(HotArticleConstants.HOT_ARTICLE_CONSUMER_QUEUE);// // 生产者队列

        return stream;

    }

    /**
     * 将传入的文章统计数据（作为字符串形式）解析为一个 ArticleVisitStreamMess 对象
     * @param
     * @param value
     * @return
     */
    private Object formatObj(String mucid, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setMuicid(mucid);
        //COMMENT:0,LIKES:0,VIEWS:0
        String[] valAry = value.split(",");
        for (String val : valAry) {
            String[] split = val.split(":");
            /*
                 COMMENT, // 评论
        LIKE, // 点赞
        VIEWS; // 浏览量（假设您想保留原有的VIEWS，表示浏览次数）
             */
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKE:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("聚合消息处理之后的结果为:{}",JSON.toJSONString(mess));
        return JSON.toJSONString(mess);
    }
}
